package timeandwindow;

import com.atguigu.pojo.Event;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

import java.time.Duration;

/**
 * Flink内置的水位线生成
 */
public class Flink02_FlinkEmbeddedWatermark {
     public static void main(String[] args) {
         Configuration conf = new Configuration();
         conf.setInteger("rest.port",5678);

         StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(conf);
             env.setParallelism(1);

             env.getConfig().setAutoWatermarkInterval(500);

             env.disableOperatorChaining();

            //对接数据源
         SingleOutputStreamOperator<Event> ds = env.socketTextStream("hadoop102", 9999)
                 .map(
                         line -> {
                             String[] fields = line.split(",");
                             return new Event(fields[0].trim(), fields[1].trim(), Long.valueOf(fields[2].trim()));
                         }
                 );

        //生成水位线
         SingleOutputStreamOperator<Event> timestampsAndWatermarks = ds.assignTimestampsAndWatermarks(
                 //1.有序流
             //  WatermarkStrategy.<Event>forMonotonousTimestamps()
             //          .withTimestampAssigner(
             //                  new SerializableTimestampAssigner<Event>() {
             //                      @Override
             //                      public long extractTimestamp(Event element, long recordTimestamp) {
             //                          return element.getTs();
             //                      }
             //                  }
             //          )
             //  //2.乱序流
             //  WatermarkStrategy.<Event>forBoundedOutOfOrderness(Duration.ofSeconds(2))//两秒延迟
             //          .withTimestampAssigner(
             //                  (element, recordTimestamp) -> element.getTs()
             //          )

                 //3.有序 + 乱序
                 WatermarkStrategy
                         //.<Event>forBoundedOutOfOrderness(Duration.ZERO)//有序
                         .<Event>forBoundedOutOfOrderness(Duration.ofSeconds(2))
                         .withTimestampAssigner(
                                 (element, ts) -> element.getTs()//返回时间戳，内部生成水位线
                         )
         );

         timestampsAndWatermarks.map( event -> event)
                 .print();


         try {
                 env.execute();
             } catch (Exception e) {
                 throw new RuntimeException(e);
             }
         }

}
